
package cn.csg.ams.collect2.poptopics.configure;// package cn.csg.ams.collect2.poptopics.kafka;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaTemplateConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;


    @Bean
    public KafkaProducer<String, String> kafkaProducer() {
        Map<String, Object> props = new HashMap<>();
        // 服务地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // key序列化
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // value序列化
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 批量处理的最大大小 byte
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 6400);
        // 重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        // acks=0 把消息发送到kafka就认为发送成功
        // acks=1 把消息发送到kafka leader分区并且写入磁盘就认为发送成功
        // acks=all 把消息发送到kafka leader分区并且leader分区的副本follower对消息进行了同步就任务发送成功
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        return new KafkaProducer<>(props);
    }


    @Bean
    public KafkaConsumer<String, String> consumerProperties() {
        Map<String, Object> props = new HashMap<>();
        // 服务地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        // latest（默认值）: 从最新的记录开始消费，即消费者将从日志的最新记录开始消
        // earliest: 从最早的记录开始消费，即消费者将从日志的最早记录开始消费。
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，
        // 会将其踢出消费组，将分区分配给别的consumer消费 ，触发rebalance .
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);

        // 是否自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 自动提交offset的时间间隔
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        // 每次poll调用中返回的最大记录数
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10);
        // session超时，超过这个时间consumer没有发送心跳, 触发 rebalance
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        // 请求超时s
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 120000);
        // key反序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // value反序列化
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 消费组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "testGroup");
        return new KafkaConsumer<>(props);
    }

}

